package dm.ass2.item;


import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


import dm.ass1.io.ProfileWritable;
import dm.ass2.ClusteringDriver;
import dm.ass2.io.CodebookUsersWritable;
import dm.ass2.item.io.ItemMapOutputValueWritable;
import dm.ass2.user.UserDriver;
import dm.ass2.user.io.CBElement;
/**
 * @author yaron
 * 
 */
public class ItemMap extends Mapper<LongWritable, Text, IntWritable, ItemMapOutputValueWritable> {
	private static final Log LOG = LogFactory.getLog(ItemMap.class);
	
	List<CodebookUsersWritable> cbUsers;
	int numOfUserClusters = 2;
	int numOfItemClusters = 2;

	@Override
	protected void map(LongWritable key, Text itemProfileString, Context context) throws IOException, InterruptedException {
		ProfileWritable itemProfile = ProfileWritable.fromString(itemProfileString.toString());
		int itemId = itemProfile.getRater();
		int[] users = itemProfile.getRated();
		int[] ratings = itemProfile.getRating();
		CBElement[] cbElements = new CBElement[numOfUserClusters];
		
		int bestClusterId  = -1; // initializing the user cluster.
		double bestClusterErr = Double.MAX_VALUE;  // initializing the minimum error.
		for (int j = 0; j < numOfItemClusters; j++) {
			double err = 0;

			for (int userIndex = 0; userIndex < users.length; userIndex++) {
				CodebookUsersWritable cbItem = findRating(users[userIndex]);
				if (cbItem == null) {
					continue;
				}
				
				double st = ratings[userIndex];
				double b = cbItem.getItemClusterRating()[j];
				err += Math.pow((st-b), 2.0);
				
				CBElement cbElement = cbElements[cbItem.getUserClusterId()];
				if (cbElement == null) {
					cbElement = new CBElement(0, 0);
					cbElements[cbItem.getUserClusterId()] = cbElement;
				}
				cbElement.increment();
				cbElement.addToSum(st);
			}
			
			if (err < bestClusterErr) {
				bestClusterErr = err;
				bestClusterId = j;
			}
		}
		
		IntWritable keyOut = new IntWritable(bestClusterId);
		ItemMapOutputValueWritable valueOut = new ItemMapOutputValueWritable(new int[] {itemId},cbElements);
		context.write(keyOut, valueOut);
	}

	private CodebookUsersWritable findRating(int item) {
		for (CodebookUsersWritable cbItem : cbUsers) {
			if (ArrayUtils.contains(cbItem.getUsrIds(), item)) {
				return cbItem;
			}
		}
		return null;
	}
	
	

	/**
	 * Initialization: loads the CodeBookItem to the memory.
	 */
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		
		this.numOfItemClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_ITEM_CLUSTERS));
		this.numOfUserClusters = Integer.parseInt(context.getConfiguration().get(UserDriver.NUM_OF_USER_CLUSTERS));

		Path cbUsersPath = new Path(ClusteringDriver.OUTPUT_DIR + ClusteringDriver.CB_USERS_FILENAME);
		FileSystem fs = FileSystem.get(context.getConfiguration());
		cbUsers = new ArrayList<CodebookUsersWritable>();

		FSDataInputStream in = fs.open(cbUsersPath);
		try {
			CodebookUsersWritable cbUser = new CodebookUsersWritable();
			cbUser.readFields(in);
			while (cbUser != null) {
				cbUsers.add(cbUser);
				cbUser = new CodebookUsersWritable();
				cbUser.readFields(in);				
			}
		} catch (EOFException e) {
			LOG.info(String.format("Setup: read %d records to cb item.", cbUsers.size()));
		} finally {
			in.close();
		}
	}

}

